package com.me.springcloud.rxjava;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

import rx.Observable;

public class MyRxJavaTest {

	static ConcurrentMap<String, String> TESTMAP = new ConcurrentHashMap<>();

	public static void main(String[] args) {
		Observable.interval(0, 300, TimeUnit.MILLISECONDS)//
				// .observeOn(Schedulers.io()).take(6)// take
				.doOnCompleted(() -> {
					System.out.println("doOnCompleted, thread=" + Thread.currentThread().getName());
				}) // do
				.subscribe(t -> {
					System.out.println("subs:" + t + ", thread=" + Thread.currentThread().getName());
				});
		pause(5);
	}

	public static void test1() {
		Observable.interval(0, 300, TimeUnit.MILLISECONDS)//
				// .observeOn(Schedulers.io()).take(6)// take
				.doOnCompleted(() -> {
					System.out.println("doOnCompleted, thread=" + Thread.currentThread().getName());
				}) // do
				.map(t -> {
					System.out.println("map:" + t + ", thread=" + Thread.currentThread().getName());
					if (t == 4) {
						return "x";
					}
					return null;
				})// do
				.filter(s -> s != null)// .subscribeOn(Schedulers.newThread()) // 订阅线程池
				.subscribe(t -> {
					System.out.println("subs:" + t + ", thread=" + Thread.currentThread().getName());
				});// sub
		pause(5);
	}

	public static void pause(int n) {
		try {
			TimeUnit.SECONDS.sleep(n);
		} catch (InterruptedException e) {
		}
	}

}
